package com.sam.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author: wm.xue
 * @date: 2021/11/29 14:10
 * @description:
 *  消费者端：位移
 *  broker: 偏移量
 */
public class _08_Consumer_Offset {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        // 关闭位移自动提交
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, Boolean.FALSE);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);

        List<PartitionInfo> partitions = consumer.partitionsFor("tp_offse_01");

        List<TopicPartition> tps = partitions.stream().map(info -> new TopicPartition(info.topic(), info.partition())).collect(Collectors.toList());
        tps.forEach(tp -> {
            consumer.seek(tp, tp.partition());
        });
        int count = 0;

        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));
                if (!records.isEmpty()) {
                    for (ConsumerRecord<String, String> record : records) {
                        Boolean res = handler(record.value());
                        if (res) {
                            offsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1));

                            if (count % 10 == 0) {
                                consumer.commitAsync(offsets, (offsets1, exception) -> {
                                    if(exception != null){
                                        System.err.println("处理异常");
                                    }
                                });

                                // 记录到数据库中
                                offsets.forEach((k,v)->{

                                });
                            }

                            count++;
                        }else{
                            System.err.println("消费失败");
                        }

                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            // 最后一次提交，使用同步提交操作
            consumer.commitSync();
        }




    }

   private static Boolean handler(String value) {
        return true;
   }
}
